package archer.back.controllers

import java.io.ByteArrayInputStream
import java.net.URL
import java.sql.Timestamp
import java.util.{ Date, UUID }

import akka.Done
import archer.back.models._
import utils.{ ArcherHelper, CommonDB, EncoderEx }
import io.circe.syntax._
import io.circe.generic.auto._

import scala.concurrent.{ ExecutionContext, Future }
import cats.implicits._
import org.jsoup.Jsoup
import org.jsoup.nodes.Element
import org.xarcher.emiya.utils.{
  EmbeddedServer,
  FutureLimited,
  FutureLimitedGen,
  FutureTimeLimited
}
//import play.api.cache.AsyncCacheApi
import play.api.http.HeaderNames
import play.api.libs.ws.WSClient

import scala.annotation.tailrec
import scala.collection.JavaConverters._
import scala.util.{ Failure, Success, Try }
import com.sksamuel.elastic4s.http.ElasticDsl._
import com.sksamuel.elastic4s.circe._
import com.sksamuel.elastic4s.http.bulk.BulkResponse
import com.sksamuel.elastic4s.http.RequestSuccess
import scala.reflect.ClassTag

class ElasticService(
  commonDB: CommonDB,
  //cacheApi: AsyncCacheApi,
  futureLimitedGen: FutureLimitedGen,
  wsClient: WSClient,
  embeddedServer: EmbeddedServer)(implicit ec: ExecutionContext)
  extends ArcherHelper {

  import models.CrawlTables._
  import models.CrawlTables.profile.api._

  val dbFutureLimited = futureLimitedGen.create(8, "数据库请求池")

  trait LimitDB {
    val db: Database
    def run[T](dbio: DBIO[T]): Future[T] = {
      dbFutureLimited.limit(() => db.run(dbio), 1, "数据库操作")
    }
  }
  val limitDB: LimitDB = new LimitDB {
    override val db = commonDB.db
  }

  /*@tailrec
  private def getPathGen(dom: Element, paths: List[Int]): List[Int] = {
    Option(dom.parent()) match {
      case Some(s) =>
        s.children().asScala.toList.zipWithIndex.find(s => s._1 == dom) match {
          case Some((_, index)) =>
            getPathGen(s, index :: paths)
          case _ =>
            throw new Exception("找不到本节点的 index")
        }
      case _ =>
        paths
    }
  }

  def getPath(dom: Element): List[Int] = {
    getPathGen(dom, List.empty)
  }

  protected def indexUrl(url: String,
                         referer: String,
                         plan: CrawlPlanRow,
                         action: CrawlActionRow,
                         futureTimeLimited: FutureTimeLimited,
                         futureLimited: FutureLimited,
                         currentDeep: Int): Future[Option[IndexTaskWrap]] = {
    def limitFuture[T](f: () => Future[T]): Future[T] =
      futureLimited.limit(
        () => futureTimeLimited.limit(() => f(), 1, s"url:$url"),
        1,
        s"url:$url")

    val hostOpt = Try {
      new URL(url).getHost
    }.toOption.filterNot(_.isEmpty)

    val protocolOpt = Try {
      new URL(url).getProtocol
    }.toOption.filterNot(_.isEmpty)

    val currentCache =
      CacheWrap(planId = plan.id, actionId = action.id, cacheUrl = url)

    futureLess.lift {
      /*val delayTimeOpt =
        futureLess.unlift(cacheApi.get[Long](currentCache.urlKey))
      val needFetch =
        delayTimeOpt.map(t => new Date().getTime > t).getOrElse(true)*/

      lazy val startTime = new Date()

      logger.info(s"正在抓取链接：${url}，链接 ElasticSearch id 为：${currentCache.urlKey}")

      val requestF = {
        val wsAction = limitFuture(() => {
          startTime
          import scala.concurrent.duration._
          wsClient.url(url).withHttpHeaders(HeaderNames.REFERER -> referer).get
        }).transform {
          case Success(response) =>
            Try {
              val successTime = new Date()
              futureLess.lift {
                val bodyBytes = response.bodyAsBytes
                val contentLength = bodyBytes.size
                val contentType = response.contentType

                val needFetchChildren = (for {
                  host <- hostOpt
                  protocol <- protocolOpt
                  planHost <- Try {
                    new URL(plan.rootUrl).getHost
                  }.toOption.filterNot(_.isEmpty)
                } yield {
                  (protocol
                    .startsWith("http")) && (host === planHost) && (contentType
                    .startsWith("text/html"))
                }).getOrElse(false)

                //要获取子链接就是说必然是 html，于是就必然有 body，下一步再做其他类型文件的索引
                val (cssBodyOpt, lawBodyOpt, subs) = if (needFetchChildren) {
                  val doc =
                    Jsoup.parse(new ByteArrayInputStream(bodyBytes.toArray),
                                null,
                                url)
                  val charset = doc.charset

                  val cssPrefix = "lolita-gentleman-"

                  val hrefLinks = doc
                    .select("*[href]")
                    .asScala
                    .toList
                    .map { dom =>
                      val urlOpt = Option(dom.attr("abs:href"))
                        .map(_.trim)
                        .filterNot(_.isEmpty)
                      urlOpt.map { s =>
                        val autalUrl =
                          EncoderEx.encodeURI(s, charset).getOrElse(s)
                        val newClass = cssPrefix + UUID.randomUUID().toString
                        dom.attr("href", autalUrl)
                        dom.addClass(newClass)
                        (autalUrl, getPath(dom), newClass)
                      }
                    }
                    .collect { case Some(s) => SubUrl(s._1, s._2, s._3) }
                  val srcLinks = doc
                    .select("*[src]")
                    .asScala
                    .toList
                    .map { dom =>
                      val urlOpt = Option(dom.attr("abs:src"))
                        .map(_.trim)
                        .filterNot(_.isEmpty)
                      urlOpt.map { s =>
                        val autalUrl =
                          EncoderEx.encodeURI(s, charset).getOrElse(s)
                        val newClass = cssPrefix + UUID.randomUUID().toString
                        dom.attr("src", autalUrl)
                        dom.addClass(newClass)
                        (autalUrl, getPath(dom), newClass)
                      }
                    }
                    .collect { case Some(s) => SubUrl(s._1, s._2, s._3) }

                  (Option(doc),
                   Option(bodyBytes.decodeString(doc.outputSettings.charset)),
                   (hrefLinks ::: srcLinks))
                } else {
                  (Option.empty, Option.empty, List.empty)
                }

                val indexTextOpt = lawBodyOpt
                val indexLawTextOpt = cssBodyOpt.map(_.text())
                val cssBodyOptStr = cssBodyOpt.map(_.outerHtml)

                val extMap = indexTextOpt match {
                  case Some(text) =>
                    text
                      .grouped(8000)
                      .toList
                      .foldLeft((List.empty[(String, String)], "", 0)) {
                        case ((list, str, index), each) =>
                          (((("law_body_" + index) -> (str + each)) :: list),
                           each.takeRight(200),
                           index + 1)
                      }
                      ._1
                  case _ =>
                    List.empty
                }

                val lawTextExtMap = indexLawTextOpt match {
                  case Some(text) =>
                    text
                      .grouped(8000)
                      .toList
                      .foldLeft((List.empty[(String, String)], "", 0)) {
                        case ((list, str, index), each) =>
                          (((("law_text_body_" + index) -> (str + each)) :: list),
                           each.takeRight(200),
                           index + 1)
                      }
                      ._1
                  case _ =>
                    List.empty
                }

                val common = CommonBody(
                  url = url,
                  contentType = Option(contentType),
                  body = indexTextOpt,
                  textBody = indexLawTextOpt,
                  referer = Option(referer),
                  law_body = extMap.toMap,
                  law_text_body = lawTextExtMap.toMap,
                  cssBody = cssBodyOptStr,
                  host = hostOpt,
                  protocol = protocolOpt,
                  deep = currentDeep,
                  wasteTime = successTime.getTime - startTime.getTime,
                  requestTime = new Date(),
                  contentLength = contentLength,
                  status = response.status,
                  subs = subs,
                  //needFetchedChildern = needFetchChildren,
                  //fetchedChildern = false,
                  planId = plan.id,
                  actionId = action.id
                )

                Option {
                  IndexTaskWrap(
                    cache = currentCache,
                    body = common,
                    afterIndexed = { () =>
                      logger.info(s"创建了 Id 为:${currentCache.urlKey} 的链接信息")
                    })
                }
              }
            }
          case Failure(e) =>
            logger.info(s"抓取url：${url} 发生错误", e)

            val failedTime = new Date()

            val common = CommonBody(
              url = url,
              contentType = Option.empty,
              body = Option.empty,
              law_body = Map.empty,
              textBody = Option.empty,
              referer = Option(referer),
              law_text_body = Map.empty,
              cssBody = Option.empty,
              host = hostOpt,
              protocol = protocolOpt,
              deep = currentDeep,
              wasteTime = failedTime.getTime - startTime.getTime,
              requestTime = new Date(),
              status = 1001,
              contentLength = -1,
              subs = List.empty,
              //needFetchedChildern = false,
              //fetchedChildern = false,
              planId = plan.id,
              actionId = action.id
            )

            Try {
              Future.successful {
                Option {

                  IndexTaskWrap(
                    cache = currentCache,
                    body = common,
                    afterIndexed = { () =>
                      logger.info(
                        s"尝试链接发生网络错误，但仍然创建了 Id 为:${currentCache.urlKey} 的链接信息")
                    })
                }
              }
            }
        }
        val f1 = futureLess.unlift(wsAction)
        futureLess.unlift(f1)
      } /* else {
        logger.info(s"url:${url}捕获缓存，不作后续操作")
        Option.empty
      }*/
      requestF
    }
  }*/

  def initSearch(
    plan: CrawlPlanRow,
    action: CrawlActionRow,
    futureTimeLimited: FutureTimeLimited,
    futureLimited: FutureLimited): V.ResultT[Int] = {
    etLess.lift {
      etLess.unlift {
        val cacheWrap = CacheWrap(
          planId = plan.id,
          actionId = action.id,
          cacheUrl = plan.rootUrl)
        val crawlModel = CrawlUrlsRow(
          id = cacheWrap.urlShaKey,
          refer = "http://www.baidu.com/",
          url = cacheWrap.cacheUrl,
          deep = 0,
          finishedCrawl = false,
          requestTime = new Timestamp(new Date().getTime),
          planId = cacheWrap.planId,
          actionId = cacheWrap.actionId)
        commonCatch(
          commonDB.db.run(
            CrawlUrls
              .map { s =>
                (
                  s.id,
                  s.refer,
                  s.url,
                  s.deep,
                  s.requestTime,
                  s.planId,
                  s.actionId)
              }
              .insertOrUpdate(
                (
                  crawlModel.id,
                  crawlModel.refer,
                  crawlModel.url,
                  crawlModel.deep,
                  crawlModel.requestTime,
                  crawlModel.planId,
                  crawlModel.actionId))))
      }: Int
    }
  }

  def crawlLoop(
    plan: CrawlPlanRow,
    crawlAction: CrawlActionRow,
    futureTimeLimited: FutureTimeLimited,
    futureLimited: FutureLimited): Future[Boolean] = {

    def limitFuture[T](f: () => Future[T]): Future[T] =
      futureLimited.limit(
        () => futureTimeLimited.limit(() => f(), 1, "发出请求"),
        1,
        "发出请求")

    val notCompleteAction = commonDB.db.run(
      CrawlUrls
        .filter(s =>
          (s.planId === plan.id) && (s.actionId === crawlAction.id) && (s.finishedCrawl === false))
        .take(80)
        .result)

    notCompleteAction
      .flatMap {
        case cModels =>
          if (cModels.isEmpty) {
            Future.successful(true)
          } else {
            val requestActions = cModels.map { model2 =>
              new CrawlUrlAction(
                url = model2.url,
                referer = model2.refer,
                plan = plan,
                action = crawlAction,
                futureTimeLimited = futureTimeLimited,
                futureLimited = futureLimited,
                wsClient = wsClient,
                currentDeep = model2.deep).indexUrl
              /*indexUrl(
                url = model2.url,
                referer = model2.refer,
                plan = plan,
                action = crawlAction,
                futureTimeLimited = futureTimeLimited,
                futureLimited = futureLimited,
                currentDeep = model2.deep
              )*/
            }

            val tasksF = Future
              .sequence(requestActions)
              .map(_.toList)

            val wrapsF = tasksF.map { tasks =>
              tasks.map { task =>
                ((
                  indexInto(embeddedServer.index.name, embeddedServer.typeName)
                  //elasticSearch id
                  .id(task.cache.elasticKey)
                  .doc(task.body),
                  //数据库 id
                  task.cache.urlShaKey),
                  task.afterIndexed,
                  task.subsModel.map(
                    sModel =>
                      CrawlUrls
                        .map { s =>
                          (
                            s.id,
                            s.refer,
                            s.url,
                            s.deep,
                            s.requestTime,
                            s.planId,
                            s.actionId)
                        }
                        .insertOrUpdate(
                          (
                            sModel.id,
                            sModel.refer,
                            sModel.url,
                            sModel.deep,
                            sModel.requestTime,
                            sModel.planId,
                            sModel.actionId))))
              }.unzip3
            }

            val insertActoins = wrapsF
              .flatMap {
                case (indexDefsWithDbIds, afterActions, dbios) =>
                  val (indexDefs, dbIds) = indexDefsWithDbIds.unzip

                  val actionsWraps = bulk(indexDefs)

                  val ids = indexDefs
                    .map(_.id)
                    .collect { case Some(s) => s }

                  lazy val insertNewUrlAction: DBIO[List[Int]] =
                    DBIO.sequence(dbios.flatten)
                  lazy val updateOldUrlAction: DBIO[Int] = CrawlUrls
                    .filter(s =>
                      (s.id inSetBind dbIds) &&
                        (s.planId === plan.id) && (s.actionId === crawlAction.id))
                    .map(s => s.finishedCrawl)
                    .update(true)

                  lazy val indexAction = embeddedServer.esLocalClient
                    .flatMap { client =>
                      client.execute {
                        actionsWraps
                      }
                    }
                    .flatMap {
                      case Left(errorMsg) =>
                        val idsStr = ids
                          .mkString("\n")
                        logger.info(
                          s"创建 Id 为:\n${idsStr}\n 的链接时发生错误，错误状态码：${errorMsg.status}，错误信息：${errorMsg.error}")
                        Future.successful(0)
                      case Right(r: RequestSuccess[BulkResponse]) =>
                        afterActions.map(s => s())
                        commonDB.db
                          .run(insertNewUrlAction
                            .flatMap((_: List[Int]) => updateOldUrlAction)
                            .transactionally)
                          .map { (_: Int) =>
                            indexDefs.size
                          }
                    }

                  /*lazy val tempUpdateAction = Future.sequence {
                    indexDefs.map(s => s.id).collect { case Some(s) => s }.map {
                      idKey =>
                        cacheApi.get[Long](idKey).flatMap { delayTimeOpt =>
                          val millions = delayTimeOpt
                            .map(t => t + (2 * 60 * 1000))
                            .filter(r => r > (new Date().getTime + 2 * 1000))
                            .getOrElse(new Date().getTime + (5 * 60 * 1000))
                          import scala.concurrent.duration._
                          val dTime =
                            (millions - new Date().getTime).millisecond
                          cacheApi.set(idKey, millions, dTime)
                        }
                    }
                  }*/

                  if (ids.isEmpty) {
                    //tempUpdateAction.map((_: List[Done]) => true)
                    Future.successful(true)
                  } else {
                    indexAction
                      //.flatMap((_: Int) => tempUpdateAction)
                      .map((_: Int) => true)
                  }
              }
              .andThen {
                case Success(s) =>
                  s: Boolean
                case Failure(e) =>
                  e.printStackTrace
              }
            insertActoins.flatMap { (s: Boolean) =>
              crawlLoop(plan, crawlAction, futureTimeLimited, futureLimited)
            }
          }
      }
      .transformWith {
        case Success(r) =>
          Future.successful(r)
        case Failure(err) =>
          err.printStackTrace
          crawlLoop(plan, crawlAction, futureTimeLimited, futureLimited)
      }

  }

}
